/*
 * Copyright 2011-2012 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.sabayframework.activity.impl;

import org.sabayframework.activity.ActivityContext;
import org.sabayframework.channel.Channel;
import org.sabayframework.channel.Lock;
import org.sabayframework.channel.ram.ArrayChannel;
import org.sabayframework.channel.ram.JoinChannel;
import org.sabayframework.channel.ram.LinkedListChannel;



/**
 * TODO document ForkJoin
 * 
 *  @author Alex Antonau
 * */
public abstract class ForkJoin<I,O> extends ReplicableActivity<I, O> {

	private I data;
	private boolean done;
	
	private final class JoinActivity extends StatefulActivity<Object[], O>{
		private O result;
		private Object[] input;
		@Override
		public void input(ActivityContext<Object[], O> ctx,	Lock<Object[]> inputLock) throws Exception {
			this.input = inputLock.get(0);
		}
		
		@Override
		public void process(ActivityContext<Object[], O> ctx) throws Exception {
			this.result = join(input);
			this.input = null;
		}

		@Override
		public void output(ActivityContext<Object[], O> ctx, Lock<O> outputLock){
			outputLock.set(0, result);
		}
		@Override
		public void afterExecute(ActivityContext<Object[], O> ctx) throws Exception {
			//close output if parent activity is done
			if(done) super.afterExecute(ctx);
		}
	}
	
	@Override
	public void afterExecute(ActivityContext<I, O> ctx) throws Exception {
		//do not call super to keep output open
	}
	
	@Override
	public final int getOutputSize(ActivityContext<I, O> ctx) {
		return 0;
	}
	@Override
	public final int getInputSize(ActivityContext<I, O> ctx) {
		return 1;
	}
	@Override
	public final void input(ActivityContext<I, O> ctx, Lock<I> inputLock){
		this.data = inputLock.get(0);
		done = ctx.getInput().isClosed() && ctx.getInput().size() == 0;
	}
	
	@SuppressWarnings("unchecked")
	@Override
	public final void process(ActivityContext<I, O> ctx) throws Exception {
		Object[] forkedData = fork(data);
		final int forkSize = forkedData == null? 0 : forkedData.length; 
		if(forkSize < 2){
			ctx.getExecutionQueue().execute(
					new Emit<O>(done? Boolean.TRUE : Boolean.FALSE, process(data)), 
					ctx.getOutput(), 
					ctx.getOutput());
		}else{
			Channel<O>[] outs = new Channel[forkSize];
			for(int i=0;i<forkSize;++i) {
				outs[i] = new LinkedListChannel<O>();
				Channel<I> input = new ArrayChannel<I>((I)forkedData[i]);
				input.close();
				ctx.getExecutionQueue().submit(replicate(), input, outs[i]);
			}
			JoinChannel jc = new JoinChannel(outs);
			ctx.getExecutionQueue().execute(new JoinActivity(), jc, ctx.getOutput());
		}
	}
	
	@Override
	public final void output(ActivityContext<I, O> ctx, Lock<O> outputLock){}
	
	public abstract Object[] fork(I in);
	public abstract O process(I in);
	public abstract O join(Object[] out);
}
